package io.coderyeah.ymcc.mq;

import com.alibaba.fastjson2.JSON;
import io.coderyeah.ymcc.constant.MQConstants;
import io.coderyeah.ymcc.constant.YmccConstants;
import io.coderyeah.ymcc.domain.MessageStation;
import io.coderyeah.ymcc.dto.StationInMQDTO;
import io.coderyeah.ymcc.service.IMessageStationService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

/**
 * @author lqs
 * @date 2022/10/28 20:20
 */
@Component
@RocketMQMessageListener(topic = MQConstants.MQ_COURSE_TOPIC,
        selectorExpression = MQConstants.MQ_COURSE_TAGS,
        consumerGroup = "service-consumer",
        messageModel = MessageModel.CLUSTERING)
@Slf4j

public class CourseMQConsumer implements RocketMQListener<MessageExt> {

    @Autowired
    private IMessageStationService stationService;

    @Override
    public void onMessage(MessageExt messageExt) {
        // 课程发布站内信消费者
        byte[] body = messageExt.getBody();
        StationInMQDTO mqdto = JSON.parseObject(new String(body), StationInMQDTO.class);
        Date date = new Date();
        List<Long> ids = mqdto.getIds();
        ids.forEach(id -> {
            MessageStation ms = new MessageStation();
            ms.setContent(mqdto.getContent());
            ms.setTitle(mqdto.getTitle());
            ms.setUserId(id);
            ms.setType(mqdto.getType());
            ms.setSendTime(date);
            ms.setIsread(0);
            stationService.insert(ms);
        });
        log.info("站内信发布成功：YES");
    }
}
